package de.preissler.dataimport;

import java.util.ArrayList;
import java.util.List;

import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
import me.prettyprint.cassandra.service.ThriftCfDef;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;

/**
 * Importiert csv Kurs- und xls DAX-Daten
 * @author Eric Preissler
 *
 */
public class Main
{
	public static final String KEYSPACE_NAME = "Exchange";
	public static final String SCF_QUOTES = "Quotes";
	public static final String SCF_DAX = "DAX";
	public static final String SCF_MDAX = "MDAX";
	public static final String SCF_TECDAX = "TDXP";
	public static int REPLICATION = 3;
	
	/*
	 * Nur statische Methoden, daher kein Konstruktor benötigt
	 */
	private Main() {};
	
	/**
	 * Start des Programms
	 * @param args Parameter werden ignoriert
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception
	{
		//Cluster initialisieren
		Cluster cluster = getCluster();
		
		//Pfad zu den CSV Dateien erfragen
		String csvPath = getCSVPath();
		
		//Pfad zu den XLS Dateien erfragen
		String xlsPath = getXLSPath();
		
		MyUtil.trace("Import process started!");
		
		//Erstellt das Schema der Datenbank
		createSchema(cluster);
		
		Keyspace keyspace = HFactory.createKeyspace(KEYSPACE_NAME, cluster);
		
		//csv Dateien importieren
		CSVImport csv = new CSVImport(cluster, keyspace, csvPath);
		csv.copyAllToCassandra();
		
		//xls Dateien importieren
		XLSImport xls = new XLSImport(cluster, keyspace, xlsPath);
		xls.copyAllToCassandra();
		
		//Alles sauber beenden
		cluster.getConnectionManager().shutdown();
		
		MyUtil.trace("Import process finished!");
		
	}
	
	/*
	 * Erstellt das Schema der Cassandra Datenbank
	 */
	private static void createSchema(Cluster cluster)
	{
		List<ColumnFamilyDefinition> scfs = new ArrayList<ColumnFamilyDefinition>(2);
		
		//Quotes Super Column Family
		BasicColumnFamilyDefinition quotesDefinition = new BasicColumnFamilyDefinition();
        quotesDefinition.setKeyspaceName(KEYSPACE_NAME);
        quotesDefinition.setName(SCF_QUOTES);
        quotesDefinition.setComparatorType(ComparatorType.UTF8TYPE);
        quotesDefinition.setColumnType(ColumnType.SUPER);
        quotesDefinition.setSubComparatorType(ComparatorType.UTF8TYPE);
        quotesDefinition.setGcGraceSeconds(0);
        scfs.add(new ThriftCfDef(quotesDefinition));
        
        //DAX Super Column Family
		BasicColumnFamilyDefinition definition = new BasicColumnFamilyDefinition();
		definition.setKeyspaceName(KEYSPACE_NAME);
		definition.setName(SCF_DAX);
		definition.setComparatorType(ComparatorType.UTF8TYPE);
		definition.setColumnType(ColumnType.SUPER);
		definition.setSubComparatorType(ComparatorType.UTF8TYPE);
		definition.setGcGraceSeconds(0);
        scfs.add(new ThriftCfDef(definition));
        
        //MDAX Super Column Family
		definition = new BasicColumnFamilyDefinition();
		definition.setKeyspaceName(KEYSPACE_NAME);
		definition.setName(SCF_MDAX);
		definition.setComparatorType(ComparatorType.UTF8TYPE);
		definition.setColumnType(ColumnType.SUPER);
		definition.setSubComparatorType(ComparatorType.UTF8TYPE);
		definition.setGcGraceSeconds(0);
        scfs.add(new ThriftCfDef(definition));
        
        //TecDAX Super Column Family
		definition = new BasicColumnFamilyDefinition();
		definition.setKeyspaceName(KEYSPACE_NAME);
		definition.setName(SCF_TECDAX);
		definition.setComparatorType(ComparatorType.UTF8TYPE);
		definition.setColumnType(ColumnType.SUPER);
		definition.setSubComparatorType(ComparatorType.UTF8TYPE);
		definition.setGcGraceSeconds(0);
        scfs.add(new ThriftCfDef(definition));
        
        KeyspaceDefinition keyspaceDefinition = 
            HFactory.createKeyspaceDefinition(KEYSPACE_NAME, "org.apache.cassandra.locator.SimpleStrategy",
            		REPLICATION, scfs);
                                           
        cluster.addKeyspace(keyspaceDefinition);
        MyUtil.trace("Keyspace \"" + KEYSPACE_NAME + "\" has been created!");
	}
	
	/*
	 * Liefert eine Refernz zu einem Cassandra-Cluster
	 */
	private static Cluster getCluster() throws Exception
	{
		String clusterName;
		String host;
		String port;
		
		System.out.println("In order to connect to cassandra we need some parameters...");
		
		System.out.print("clustername: ");
		clusterName = MyUtil.readLine();
		
		System.out.print("host: ");
		host = MyUtil.readLine();
		
		System.out.print("port: ");
		port = MyUtil.readLine();
		
		return HFactory.getOrCreateCluster(clusterName, host + ":" + port);
	}
	
	/*
	 * Erfagt den Pfad zu den CSV Dateien
	 */
	private static String getCSVPath() throws Exception
	{
		System.out.println("Please enter the path of the csv files:");
		
		return MyUtil.readLine();
	}
	
	/*
	 * Erfagt den Pfad zu den XLS Dateien
	 */
	private static String getXLSPath() throws Exception
	{
		System.out.println("Please enter the path of the xls files:");
		
		return MyUtil.readLine();
	}
}
